//*********************************************
//ADRIAN SALAZAR && JUSTIN MICHALE SCHWARTZBECK
//HOMEWORK 2
//FEB. 23, 2012
//
//Client Buffer
//*********************************************

#include "clientbuffer.h"
#include <stdio.h>
#include <errno.h>
#include <ctime>
#include <pthread.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <list>
#include <stdlib.h>
#include <unistd.h>
#include <arpa/inet.h>

#define UDP_PAYLOAD_LEN 65507
#define EPOCH_LEN 3
#define EPOCH_EXP 3

using namespace std;

//*********************************************
//Client IO Thread Func: This function will handle
//	all of the sending and receiving. of the 
//	packets. 
//*********************************************

void * cli_io_thread_func(void * arg) 
{
	//printf("Started Client IO thread\n");
	ClientBuffer * client = (ClientBuffer*)arg;
	
	// Time data
	time_t epoch_timer = time(NULL);
	client->last_io_event = time(NULL);
	unsigned short epoch_counter = 0;

	
	//Init needed variables
	socklen_t addr_len = sizeof(client->client);
	int a; 
	char buf[UDP_PAYLOAD_LEN];
	int rec = 0;
	int rec2 = 0;
	uint32_t temp_id; 
	uint32_t temp_seq;
	uint32_t temp_ack_seq = 1;
	int pid;				
	uint32_t recv_seq = 1;

	//Send the initial request to the server
	if(client->state == 0)
	{
		//printf("Sending Initial Request \n");
		//Create an empty packet to send
		Packet client_packet(0, 0, 0, 0);
		string client_str;
		client_packet.msg.SerializeToString(&client_str);
		a = sendto(client->sock, client_str.c_str(), client_str.size(), 0,(struct sockaddr*)&client->client, addr_len);
		
		//Lock variables to prevent any race conditions.
		pthread_mutex_lock(&client->timer_lock);
		client->last_io_event = time(NULL);
		client->state = 1;
		pthread_mutex_unlock(&client->timer_lock);

	}
	
	
	//Continue to listen and send while the client is running
	while(client->running)
	{
		
		//Set up listening variables
		fd_set fds;
		FD_ZERO(&fds);
		FD_SET(client->sock, &fds);
		
		//set up time variables
		struct timeval timeout;
		timeout.tv_usec = 100;
		timeout.tv_sec = 0;	
		
		time_t seconds;
		time(&seconds);
		srand((unsigned int) seconds);
		int rand_drop = rand() % 100;
		
		//Check to see if a packet will be dropped or not
		if(rand_drop >= (client->drop_rate * 100))
		{
		
			//Check to see if there is something to read from the socket
			int ready = select(client->sock+1, &fds, 0, 0, &timeout);
			if(ready > 0)
			{
				//Receive from the server
				a = recvfrom(client->sock, buf, UDP_PAYLOAD_LEN, 0, NULL, NULL);
				Packet p((uint8_t*)buf, a);
				temp_id= p.get_conn_id();
				temp_seq =  p.get_seq_num();
				//convert the data to seq and conn id
				//printf("conn id: %d. seq: %d. payload: %s\n", temp_id, temp_seq, p.get_payload());
				
				//Check to see if there are any messages that are needing
				//an acknowledgement
				pthread_mutex_lock(&client->list_lock);
				if(client->ack_buf.empty() == false)
				{	
					Packet temp_ack = client->ack_buf.front();
					temp_ack_seq = temp_ack.get_seq_num();
					//printf("Expecting Ack: %i\n", temp_ack_seq);
				
				}
				else
				{
					//printf("ack is -1\n");
					//temp_ack_seq = -1;
				}
				
				pthread_mutex_unlock(&client->list_lock);
				
				//Ready to receive initial acknowledgement form server
				if(temp_seq == 0 && client->state==1)
				{
					//printf("case 1\n");
					//Initial acknowledgment
					pthread_mutex_lock(&client->list_lock);
					//Set needed variables
					client->state = 2; 
					client->conn_id = temp_id;
					client->seq_num = temp_seq;
					client->ack_seq = temp_seq;
					temp_ack_seq = temp_seq;
					pthread_mutex_unlock(&client->list_lock);
						
					//IO Even occurred
					pthread_mutex_lock(&client->timer_lock);
					client->last_io_event = time(NULL);
					epoch_counter = 0;
					pthread_mutex_unlock(&client->timer_lock);
				}
				else if(client->state == 2 && temp_ack_seq == temp_seq && strlen((const char *)p.get_payload() ) == 0) 
				{	
					rec2++;
					//printf("case 2\n");
					//Receive the acknowledgement from the expected packet 
					//printf("Receive Ack Seq # %d\n", temp_seq);
					pthread_mutex_lock(&client->list_lock);
					client->ack_seq = temp_seq;
					
					//Only take an item out if the list is not empty
					if(client->ack_buf.empty() == false)
						client->ack_buf.pop_front();
					pthread_mutex_unlock(&client->list_lock);
					
					//IO Event occurred
					pthread_mutex_lock(&client->timer_lock);
					client->last_io_event = time(NULL);
					epoch_counter = 0;
					pthread_mutex_unlock(&client->timer_lock);
			
				}
				else if(client->state == 2 && temp_seq == recv_seq && strlen((const char *)p.get_payload() ) > 0)
				{
					//printf("case 3\n");
					//Received a Packet from the server
					rec++;
					//printf("Receive packet Seq # %d %d\n", temp_seq, a);
					//Receive Packet, put it in list to read from.
					pthread_mutex_lock(&client->list_lock);
					Packet tmp_packet((uint8_t*)buf, a);
					client->in_buf.push_back(tmp_packet);
					
					//Create response packet
					Packet ack_pack(client->conn_id, temp_seq, 0, 0);
					string ack_str;
					ack_pack.msg.SerializeToString(&ack_str);

					//send acknowledgment
					sendto(client->sock, ack_str.c_str(), ack_str.size(), 0, (struct sockaddr*)&client->client, addr_len);
					pthread_mutex_unlock(&client->list_lock);
					
					//Increase sequence number
					recv_seq++;
		
					//Reset epoch timer
					pthread_mutex_lock(&client->timer_lock);
					client->last_io_event = time(NULL);
					epoch_counter = 0;
					pthread_mutex_unlock(&client->timer_lock);

				}
				else
				{
					//An out of sequence message was received
					//will ignore it.
					//printf("case 4\n");
					//printf("Bad Message Seq # %d %d\n", temp_seq, a);
					//Reset epoch timer
					pthread_mutex_lock(&client->timer_lock);
					client->last_io_event = time(NULL);
					epoch_counter = 0;
					pthread_mutex_unlock(&client->timer_lock);
				}
				
			}
		
		}
			//Check to see that there is an item in the outgoing list. 
			if(client->out_buf.empty() == false && client->state == 2)
			{
			
				//get the packet and remove from list
				pthread_mutex_lock(&client->list_lock);
				//send packet to server
				Packet client_packet = client->out_buf.front();
				string client_str;
				client_packet.msg.SerializeToString(&client_str);
				
				//printf("Sending Packet with Seq # %d \n", client_packet.get_seq_num());
				a = sendto(client->sock, client_str.c_str(), client_str.size(), 0, (struct sockaddr*)&client->client, addr_len);
				client->ack_buf.push_back(client_packet);
				client->out_buf.pop_front();
				pthread_mutex_unlock(&client->list_lock);
				
				//Reset epoch timer
				pthread_mutex_lock(&client->timer_lock);
				client->last_io_event = time(NULL);
				epoch_counter = 0;
				pthread_mutex_unlock(&client->timer_lock);
				
			}
			
		// Handle epoch events
		time_t current_time = time(NULL);
		if(current_time - epoch_timer >= client->epoch) 
		{
			//One epoch has passed
			//printf("One epoch has passed\n");
			if(client->state == 1)
			{
				//printf("Resending New Connection Request\n");
				//send new connection request
				Packet tmp_packet(0, 0, 0, 0);
				string tmp_str;
				tmp_packet.msg.SerializeToString(&tmp_str);
				sendto(client->sock, tmp_str.c_str(), tmp_str.size(), 0,(struct sockaddr*)&client->client, addr_len);
				//client->last_io_event = time(NULL);
			}
			if(client->state == 2)
			{	
				// Acknowledge latest received packet
				//printf("Resending Last Ack ");
				if(rec2 > 0)
				{
					//Send a new packet if no data has been received
					Packet tmp_packet(client->conn_id, client->ack_seq, 0, 0);
					string tmp_str;
					tmp_packet.msg.SerializeToString(&tmp_str);
					sendto(client->sock, tmp_str.c_str(), tmp_str.size(), 0,(struct sockaddr*)&client->client, addr_len);
					//printf(" rec > 0: %i\n", tmp_packet.get_seq_num());
				}
				else
				{
					//Otherwise acknowledge the last packet received
					Packet tmp_packet(client->conn_id, 0, 0, 0);
					string tmp_str;
					tmp_packet.msg.SerializeToString(&tmp_str);
					sendto(client->sock, tmp_str.c_str(), tmp_str.size(), 0,(struct sockaddr*)&client->client, addr_len);
					//printf(" rec = 0: %i\n", tmp_packet.get_seq_num());
				}	
				
			}
		
			if(client->ack_buf.empty()==false)
			{
				//Resend packet if acknowledgement has not been received
				//printf("Resending Last Message\n");
				Packet tmp_packet = client->ack_buf.front();string tmp_str;
				tmp_packet.msg.SerializeToString(&tmp_str);
				sendto(client->sock, tmp_str.c_str(), tmp_str.size(), 0,(struct sockaddr*)&client->client, addr_len);
				//client->last_io_event = time(NULL);
			}
			
			pthread_mutex_lock(&client->timer_lock);
			epoch_timer = time(NULL);
			pthread_mutex_unlock(&client->timer_lock);
			
		}
		
		// Handle timeouts
		if(current_time - client->last_io_event >= client->epoch) 
		{
			pthread_mutex_lock(&client->timer_lock);
			epoch_counter++;
			client->last_io_event = time(NULL);

			pthread_mutex_unlock(&client->timer_lock);
			//printf("%d inactive epochs\n", epoch_counter);
		}
		if(epoch_counter == 3) {
			// Connection has timed out
			client->running = false;
			client->finish();
		}
	}
	//printf("Shutting down client. \n");
}

//*********************************************
//Create Instance of Class:
//	Set up connection to the server. The server needs to exist
//	otherwise the creation will fail.
//	Class will take default arguments from the lsp.c for the
//	epoch and drop rate.
//*********************************************
ClientBuffer::ClientBuffer(const char * ip, int in_port, double epoch_len, int e_count, double d_rate) 
{

	//set up default variables
	port = in_port;
	ack_seq = 0; 
	state = 0;
	
	//Set up connection
	client.sin_addr.s_addr = inet_addr(ip);
	client.sin_port = htons(port);
	client.sin_family = AF_INET;

	
	epoch = epoch_len;
	numepochs = e_count;
	drop_rate = d_rate;

	sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
	if(sock == -1) 
	{
		state = errno;
		errmsg = "Failed to create socket";
		return;
	}
	
	int result;

	running = true;
	
	//Spawn a new thread
	result = pthread_create(&io_thread, 0, cli_io_thread_func, this);
	pthread_mutex_init(&list_lock, NULL);
	pthread_mutex_init(&in_lock, NULL);
	pthread_mutex_init(&out_lock, NULL);
	pthread_mutex_init(&timer_lock, NULL);
	if(result == -1)
	{
		state = errno;
		errmsg = "Failed to spawn I/O thread";
		return;
	}
}


//*********************************************
//Return whether the server is running or not.
//*********************************************
bool ClientBuffer::is_running() 
{
	return running;
}

//*********************************************
//Terminate the the thread and close the connection
//to the server
//*********************************************
void ClientBuffer::finish() 
{
	state = -1;
	running = false;
	pthread_join(io_thread, NULL);
	close(sock);
}

//*********************************************
//Read data from the incoming buffer. 
//	Sleep while the buffer is empy
//	or return in failure if the 
//	connection was lost.
//*********************************************
int ClientBuffer::read(uint8_t * data)
{
	//Sleep while there is nothing in buffer	
	while(in_buf.empty() == true && running == true) sleep(.001);
	
	//Return if connection lost
	if(running == false)
		return -1;
	else
	{
		//Read from the list
		pthread_mutex_lock(&list_lock);
		Packet tmp_packet = in_buf.front();
		memcpy(data, tmp_packet.get_payload(), strlen((const char *)tmp_packet.get_payload()));
		in_buf.pop_front();
		pthread_mutex_unlock(&list_lock);

		return strlen((const char *)tmp_packet.get_payload());
	}
	
}
//*********************************************
//Write: Place data to be sent in the list.
//	only execute while the server is connected
//*********************************************
int ClientBuffer::write(uint8_t * data, uint32_t len)
{
	while(state != 2 && running == true) sleep(.001); 

	if(running == false)
		return -1;
	else
	{
		pthread_mutex_lock(&list_lock);
		Packet tmp_packet(conn_id, ++seq_num, data, len);
		out_buf.push_back(tmp_packet);
		pthread_mutex_unlock(&list_lock);
		return 1;
	}
	
}
